
package hasthi.coordinator;

import static hasthi.common.constants.WSDMConstants.RESOURCE_STATUS_PARAM;
import hasthi.actuators.ActionCenter;
import hasthi.actuators.ActionContext;
import hasthi.actuators.ManagementAction;
import hasthi.common.ChangeListener;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.SafeScheduler;
import hasthi.common.Utils;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.WSDMConstants.SystemState;
import hasthi.decision.RuleEngine;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerContext;
import hasthi.manager.ResourceOutline;
import hasthi.manager.JoinManagementRingResponseDocument.JoinManagementRingResponse;
import hasthi.manager.ManagementRingDocument.ManagementRing;
import hasthi.manager.ManagementRingDocument.ManagementRing.ManagerData;
import hasthi.manager.ManagerHeartBeatResponseDocument.ManagerHeartBeatResponse;
import hasthi.metamodel.SummerizedResourceModel;
import hasthi.metamodel.resource.ManagedResource;
import hasthi.metamodel.resource.ResourceModelContext;
import hasthi.metamodel.utils.SystemHealthState;
import hasthi.tools.systemstat.SystemStat;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlObject;


/**
 * Keep track of the states of the coordinaotr
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class CoordinatorContext {
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    private Map<QName,RemoteManagerContext> resourceAssignmentMap = new Hashtable<QName, RemoteManagerContext>();
    private Map<String,RemoteManagerContext> manager2ResourcesMap = new  ConcurrentHashMap<String, RemoteManagerContext>();
    private SortedSet<RemoteManagerContext> sortedManagers = new TreeSet<RemoteManagerContext>();
    
    private ManageMeMessageBag manageMeMessageBag;
    
    private List<ManagerAddress> sucessorList = new ArrayList<ManagerAddress>();
    private SummerizedResourceModel resourceModel;
    /**
     * Manager Context for the Coordinator
     */
    private ManagerContext myManagerContext;
    private int maxRankKnown = 1;
    private Map<String, String> rankMap = new HashMap<String, String>();
    
    private RuleEngine ruleEngine;
    /**
     * This parameter keep track of the ManagerContext of the coordinator. Coordinator does not assign 
     * resources to itself unless there is no other manager found
     */
    private RemoteManagerContext localManagerContext;
    private DistributedLogger dlogger;
    
    private CoordinationDemon coordinationDemon;
    
    private AtomicInteger resourceCount = new AtomicInteger();
    private final CoordinatorMessageProcessor coordinatorInstance; 
    private final SystemStat systemStat = new SystemStat();
    private Map<Long,QName> resource2Remove = new TreeMap<Long, QName>();
    private boolean activeManagement = true;
    
    public CoordinatorContext(ManagerContext myManagerContext) throws HasthiException {
        this.myManagerContext = myManagerContext;
        ResourceModelContext resourceModelContext = new ResourceModelContext(this,new ResourceChangeListener());
        resourceModel = new SummerizedResourceModel(resourceModelContext);
        //By default coordinator has rank 0
        localManagerContext = new RemoteManagerContext(myManagerContext.getMyAddress(),0);
        String globalRuleFile = myManagerContext.getBaseContext().getGlobalRuleSet();
        log.info("Using Global rules from "+ globalRuleFile);
        
        //Initialize the Action Center
        ActionContext actionContext = new ActionContext(myManagerContext.getBaseContext(),true);
        actionContext.setCoordinatorContext(this);
        actionContext.setResourceModel(resourceModel);
        actionContext.setUnicastContainer(myManagerContext.getUnicastContainer());
        actionContext.setMulticastContainer(myManagerContext.getMulticastContianer());
        actionContext.setScheduler(myManagerContext.getScheduler());
        ActionCenter actionCenter = myManagerContext.getBaseContext().createActuatorBase();
        actionCenter.init(actionContext);
        ruleEngine = new RuleEngine(globalRuleFile,actionCenter);
        actionContext.setRuleEngine(ruleEngine);
        

        dlogger = myManagerContext.getBaseContext().getDlogger();
        
        //Add myself as a eligible Manager
        String id = Utils.mngAdr2Id(myManagerContext.getMyAddress());
        manager2ResourcesMap.put(id, localManagerContext);
        sortedManagers.add(localManagerContext);
        maxRankKnown = myManagerContext.getMyRank();
        coordinatorInstance = new CoordinatorMessageProcessor(this,myManagerContext);
        systemStat.setCoordiantorAddress(getMyManagerContext().getMyAddress());
        
        manageMeMessageBag = new ManageMeMessageBag(this);
    }

    /**
     * This list is sorted by the rank
     * @return
     */
    public synchronized List<ManagerAddress> createSucessorList(){
        sucessorList.clear();
        int sucessorCout = 0;
        for (RemoteManagerContext managerContext:sortedManagers) {
            if (sucessorCout < myManagerContext.getBaseContext().getSucessorCount() &&
                    !Utils.equals(managerContext.getAddress(), myManagerContext.getMyAddress())) {
                sucessorList.add(managerContext.getAddress());
                sucessorCout++;
            }
        }
        return sucessorList;
    }
    
    
    public synchronized void copySucessors(XmlObject response){
        if(response instanceof ManagerHeartBeatResponse){
            ManagerHeartBeatResponse heartBeatResponse = (ManagerHeartBeatResponse)response;
            for(ManagerAddress address:sucessorList){
                ManagerAddress sucessor = heartBeatResponse.addNewSucessor();
                sucessor.setSoapAddress(address.getSoapAddress()); 
                sucessor.setP2Paddress(address.getP2Paddress());
                sucessor.setRank(address.getRank());
            }
        }else{
            JoinManagementRingResponse heartBeatResponse = (JoinManagementRingResponse)response;
            for(ManagerAddress address:sucessorList){
                ManagerAddress sucessor = heartBeatResponse.addNewSucessor();
                sucessor.setSoapAddress(address.getSoapAddress()); 
                sucessor.setP2Paddress(address.getP2Paddress());
                sucessor.setRank(address.getRank());
            }
        }
         
        
    }
//    public synchronized List<ManagerAddress> getSucessorList(){
//        return sucessorList;
//    }
    
    public RemoteManagerContext getManagerForResource(QName resourceName){
        return resourceAssignmentMap.get(resourceName);
    }
    
    public RemoteManagerContext getManagerData(String id){
        return manager2ResourcesMap.get(id);
    }
    
    public synchronized int addNewManager(ManagerAddress address){
        int newRank = address.getRank();
        if(newRank > 0 && newRank != Integer.MAX_VALUE 
                && !rankMap.containsKey(String.valueOf(newRank))){
            rankMap.put(String.valueOf(newRank), String.valueOf(newRank));
            if(newRank > maxRankKnown){
                maxRankKnown = newRank;
            }
        }else{
            newRank = maxRankKnown + 1;
            maxRankKnown = newRank;
            rankMap.put(String.valueOf(newRank), String.valueOf(newRank));
        }
        address.setRank(newRank);
        RemoteManagerContext managerContext = new RemoteManagerContext(address,newRank); 
        String id = Utils.mngAdr2Id(address);
        manager2ResourcesMap.put(id,managerContext);
        sortedManagers.add(managerContext);
        dlogger.info("New Manager "+ address.getSoapAddress() + "("+newRank+") joined");
        return newRank;
    }
    
    
    public synchronized void removeManager(ManagerAddress address){
        String id =  Utils.mngAdr2Id(address);
        RemoteManagerContext context = manager2ResourcesMap.remove(id);
        if(context != null){
            sortedManagers.remove(context);    
        }else{
            dlogger.info("UNEXPECTED CONDITION:can not remove manager " + id + " from "+ manager2ResourcesMap.keySet() );
        }
        dlogger.info("Manager "+ address.getSoapAddress() + " Left");
    }
    
    




    public SummerizedResourceModel getResourceModel() {
        return resourceModel;
    }
    
    
    /**
     * Figure out best Manager to assign this resource, assgin it and return the address
     * This does not add the resource
     * @param resourceName
     * @return
     * @throws HasthiException
     */
    public RemoteManagerContext findBestManager4Resource(QName resourceName) throws HasthiException{
        RemoteManagerContext resourceCaretaker = null;
        int minAssignedCount = Integer.MAX_VALUE;
        for(RemoteManagerContext manager:manager2ResourcesMap.values()){
            if(manager != localManagerContext 
                    &&  manager.getAssignedResources().size() < minAssignedCount){
                minAssignedCount = manager.getAssignedResources().size();
                resourceCaretaker = manager;
            }
        }
        
        /**
         * If no manager found, then assign it to himself
         */
        if(resourceCaretaker == null){
            resourceCaretaker = localManagerContext; 
        }
        return resourceCaretaker;
    }
    
    public void addResource2Manager(QName resourceName,RemoteManagerContext resourceCaretaker){
        resourceAssignmentMap.put(resourceName, resourceCaretaker);
        resourceCaretaker.addResource(resourceName);
        resourceCount.incrementAndGet();
        dlogger.debug("Assigment "+ resourceName.getLocalPart()  + "->"+ resourceCaretaker.getAddress().getSoapAddress());
    }
    
    public void removeResource(QName resourceName){
        RemoteManagerContext resourceCaretaker = resourceAssignmentMap.remove(resourceName);
        resourceCaretaker.removeResource(resourceName);
        resourceCount.addAndGet(-1);
        dlogger.info("Remove Assigment "+ resourceName.getLocalPart()  + "->"+ resourceCaretaker.getAddress().getSoapAddress());
    }
    
    public ManagerContext getMyManagerContext() {
        return myManagerContext;
    }

    public RuleEngine getRuleEngine() {
        return ruleEngine;
    }
    public synchronized int getManagerCount(){
        return sortedManagers.size();
    }
    
    public synchronized ManagementRing toXml() throws HasthiException{
        ManagementRing ringData = ManagementRing.Factory.newInstance();
        ringData.setCoordinator(myManagerContext.getMyAddress());
        ringData.setActiveManagement(isActiveManagement());
        SystemHealthState systemHealthState = (SystemHealthState)ruleEngine.getActionCenter().get(SystemHealthState.SYSTEM_HEALTH_STATE_KEY);
        if(systemHealthState != null){
            ringData.setSystemHealth(systemHealthState.isSystemHealthy()?"Healthy":"Faulty");
        }
        for(RemoteManagerContext managerData:sortedManagers){
            if(!myManagerContext.getMyAddress().getSoapAddress().equals(managerData.getAddress().getSoapAddress())
                    || managerData.getAssignedResources().size() > 0){
                ManagerData managerDataX = ringData.addNewManagerData();
                managerDataX.setManagerAddress(managerData.getAddress());
                
                List<QName> assignedResources = managerData.getAssignedResources();
                for(QName name:assignedResources){
                    ManagedResource resource = resourceModel.getResource(name);
                    if(resource != null){
                        ResourceOutline resourceOutline = managerDataX.addNewResourceData();
                        resourceOutline.setName(name.toString());
                        resourceOutline.setState(resource.getState());    
                    }
                }
            }
        }
        return ringData;
    }

    public synchronized CoordinationDemon getCoordinationDemon() {
        if(coordinationDemon == null){
            coordinationDemon =  new CoordinationDemon(this);
        }
        return coordinationDemon;
    }
    
    public int getResourceCount(){
        return resourceCount.get();
    }
    
    public Collection<RemoteManagerContext> getManagers(){
        return manager2ResourcesMap.values();
    }

    public CoordinatorMessageProcessor getCoordinatorMessageProcessor() {
        return coordinatorInstance;
    }
    
    public long getAge(){
        long coordinatorTimeStamp = myManagerContext.getCoordinatorTimeStamp();
        if(coordinatorTimeStamp == -1){
            return 0;
        }else{
            long coordinatorAgeSteps = (System.currentTimeMillis() -
                    coordinatorTimeStamp)/(myManagerContext.getBaseContext().getCoordinatorEpochSeconds()*1000);
            return coordinatorAgeSteps;
        }
    }

    public SystemStat getSystemStat() {
        return systemStat;
    }
    
    
    public class ResourceChangeListener implements ChangeListener{
        public void propertyAdded(QName resourceName, QName propertyName) {
        }

        public void propertyChanged(QName resourceName, QName propertyName, Object newValue) {
            if(RESOURCE_STATUS_PARAM.getLocalPart().equals(propertyName.getLocalPart())){
                long key = -1;
                if(newValue.equals(SystemState.CrashedState.toString())) {
                    key = System.currentTimeMillis() + myManagerContext.getBaseContext().getManagerEpochSeconds()*1000*3;
                }else if(newValue.equals(SystemState.FaultyState.toString())){
                    key = System.currentTimeMillis() + myManagerContext.getBaseContext().getManagerEpochSeconds()*1000*10;
                }else if(newValue.equals(SystemState.StoppedState.toString())){
                    key = System.currentTimeMillis() + myManagerContext.getBaseContext().getManagerEpochSeconds()*1000*2;
                }else if(newValue.equals(SystemState.RepairedState.toString())){
                    key = System.currentTimeMillis() + myManagerContext.getBaseContext().getManagerEpochSeconds()*1000*5;                    
                }
                if(key > 0){
                    synchronized (resource2Remove) {
                        QName name = resource2Remove.put(key ,resourceName);
                        while(name != null){
                            key++;
                            name = resource2Remove.put(key,resourceName);
                        }
                    }
                }
            }
        }

        public void propertyRemoved(QName resourceName, QName propertyName) {
        }

        public void resourceAdded(QName resourceName) {
        }

        public void resourceRemoved(QName resourceName) {
        }
    }


    public Map<Long, QName> getResource2Remove() {
        return resource2Remove;
    }

    public ManageMeMessageBag getManageMeMessageBag() {
        return manageMeMessageBag;
    }

    public boolean isActiveManagement() {
        return activeManagement;
    }

    public void setActiveManagement(boolean activeManagement) {
        this.activeManagement = activeManagement;
    }
    
}

